package tech.realcpf.flink.catalog.store;

import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.options.GetOption;
import io.vertx.core.json.Json;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.catalog.AbstractCatalogStore;
import org.apache.flink.table.catalog.CatalogDescriptor;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class JcStore extends AbstractCatalogStore {
    private static final Logger LOG = LoggerFactory.getLogger(JcStore.class);
    private final Client etcdClient;

    private final String searchPrefix;

    public JcStore(String searchPrefix,String endPoints,String user,String passwd) {
        this.etcdClient = Client.builder()
                                        .user(ByteSequence.from(user,StandardCharsets.UTF_8))
                                        .password(ByteSequence.from(passwd,StandardCharsets.UTF_8))
                                        .endpoints(endPoints.split(",")).build();
        this.searchPrefix = searchPrefix;
    }

    @Override
    public void storeCatalog(String s, CatalogDescriptor catalogDescriptor) throws CatalogException {
        String data = Json.encode(catalogDescriptor.getConfiguration().toMap());
        try {
            etcdClient.getKVClient().put(ByteSequence.from(String.format("%s%s",searchPrefix,s),StandardCharsets.UTF_8),
                    ByteSequence.from(data,StandardCharsets.UTF_8)).get(1L,TimeUnit.SECONDS);
        } catch (Exception e) {
            LOG.error("error when store catalog {}",s,e);
        }
    }

    @Override
    public void removeCatalog(String s, boolean b) throws CatalogException {
        try {
            etcdClient.getKVClient().delete(ByteSequence.from(String.format("%s%s",searchPrefix,s),StandardCharsets.UTF_8)).get(1L,TimeUnit.SECONDS);
        } catch (Exception e) {
            LOG.error("error when remove catalog {}",s,e);
        }
    }

    @Override
    public Optional<CatalogDescriptor> getCatalog(String s) throws CatalogException {
        try {
            GetResponse response = etcdClient.getKVClient().get(ByteSequence.from(String.format("%s%s",searchPrefix,s), Charset.defaultCharset()),
                    GetOption.builder().isPrefix(Boolean.FALSE).build()).get(1L, TimeUnit.SECONDS);
            if (response.getCount() > 0) {
                String res = response.getKvs().get(0).getValue().toString(StandardCharsets.UTF_8);
                Map<String,String> data = Json.decodeValue(res, Map.class);
                return Optional.of(CatalogDescriptor.of(s,Configuration.fromMap(data)));
            }
            return Optional.empty();
        } catch (Exception e) {
            LOG.error("error when get catalog {}",s,e);
        }

        return Optional.empty();
    }

    @Override
    public Set<String> listCatalogs() throws CatalogException {
        try{
            GetResponse response = etcdClient.getKVClient().get(ByteSequence.from(searchPrefix, Charset.defaultCharset()),
                    GetOption.builder().isPrefix(Boolean.TRUE).build()).get(1L, TimeUnit.SECONDS);
            return response.getKvs().stream().map(KeyValue::getKey).map(ByteSequence::toString)
                    .map(s-> s.replaceFirst(searchPrefix,"")).collect(Collectors.toSet());
        }catch (Exception e) {
            LOG.error("error when list catalog",e);
        }
        return Set.of();
    }

    @Override
    public boolean contains(String s) throws CatalogException {
        return listCatalogs().contains(s);
    }

    @Override
    public void close() {
        if (Objects.nonNull(etcdClient)) {
            etcdClient.close();
        }
    }
}